package io.awen.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import io.awen.modules.sign.entity.SignCarEntity;
import io.awen.modules.sign.entity.SignCarLonLatEntity;
import io.awen.modules.sign.entity.SignContainerEntity;
import io.awen.modules.sign.service.SignCarService;
import io.awen.modules.sign.service.SignContainerService;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
public class MqttPushClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);

    @Autowired
    private SignCarService signCarService;
    @Autowired
    private SignContainerService signContainerService;

    private static MqttClient client;

    private static MqttClient getClient() {
        return client;
    }

    private static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    /**
     * 客户端连接
     *
     * @param host      ip+端口
     * @param clientID  客户端Id
     * @param username  用户名
     * @param password  密码
     * @param timeout   超时时间
     * @param keepalive 保留数
     */
    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive, String[] mqttTopic) {
        MqttClient client;
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            // 设置自动重连
            options.setAutomaticReconnect(true);
            MqttPushClient.setClient(client);
            try {
                // 此处使用的MqttCallbackExtended类而不是MqttCallback，是因为如果emq服务出现异常导致客户端断开连接后，重连后会自动调用connectComplete方法
                client.setCallback(new MqttCallbackExtended() {
                    @Override
                    public void connectComplete(boolean reconnect, String serverURI) {
//                        System.out.println("连接完成...");
                        try {
                            // 重连后要自己重新订阅topic，这样emq服务发的消息才能重新接收到，不然的话，断开后客户端只是重新连接了服务，并没有自动订阅，导致接收不到消息
                            client.subscribe(mqttTopic);
//                            System.out.println("订阅成功..." + mqttTopic);
                        } catch (Exception e) {
                            logger.info("订阅出现异常:{}", e);
                            System.out.println("订阅出现异常..." + e);
                        }
                    }

                    @Override
                    public void connectionLost(Throwable cause) {
//                        System.out.println("失去连接....");
//                        cause.printStackTrace();
                    }

                    @Override
                    public void messageArrived(String topic, MqttMessage message) throws Exception {
                        // subscribe后得到的消息会执行到这里面
                        String content = new String(message.getPayload());
                        System.out.println("接收消息主题 : " + topic);
                        System.out.println("接收消息Qos : " + message.getQos());
                        System.out.println("接收消息内容 : " + content);
                        // 处理数据 messageConsumeService.dealEmqContent(content);
//                        System.out.println("mqtt-message：" + message);
//                        logger.info("mqtt-message：" + message);
//=================================================================================================================================================
//                        if ("datachange_S_KIO_Project".equals(topic)) { //液位对接
//                            Map<String, String> jsonMap = JSON.parseObject(content, new TypeReference<HashMap<String, String>>() {
//                            });
//                            JSONArray array = JSONArray.parseArray(jsonMap.get("Objs"));
//                            String pVs = jsonMap.get("PVs");
//                            JSONObject jsonObject = JSON.parseObject(pVs);
//
//                            Map o = (Map) array.get(0);
//                            String n = o.get("N").toString();
//                            Double level = Double.parseDouble(jsonObject.get("1").toString());
//
//                            SignContainerEntity container = signContainerService.getBySensorId(n);
//
//                            if (container != null) {
//                                signContainerService.updateSensorValue(n, level, new Date());
//                            }
//                        } else if ("topic-jtt808".equals(topic)) { //对接车辆
//                            if (!StringUtils.isBlank(content)) {
////                                System.out.println("接收到的消息： " + content);
//                                Map<String, String> jsonMap = JSON.parseObject(content, new TypeReference<HashMap<String, String>>() {
//                                });
//                                String clientId = jsonMap.get("clientId");
//                                Double direction = Double.parseDouble(jsonMap.get("direction"));
//                                Double latitude = Double.parseDouble(jsonMap.get("latitude")) / 1000000;
//                                Double longitude = Double.parseDouble(jsonMap.get("longitude")) / 1000000;
//                                Double speed = Double.parseDouble(jsonMap.get("speed")) / 10;
//
//                                String weight = null;
//                                Double deformation = null;
//                                Date createTime = null;
//                                try {
//                                    String attributes = jsonMap.get("attributes");
//                                    JSONObject attributesJson = JSON.parseObject(attributes);
//                                    String LoadInfo = attributesJson.get("235").toString();
//                                    JSONObject LoadInfoJson = JSON.parseObject(LoadInfo);
//                                    String LoadWeight = LoadInfoJson.get("loadWeight").toString();
//                                    JSONObject LoadWeightJson = JSON.parseObject(LoadWeight);
//                                    weight = LoadWeightJson.get("250").toString();
//                                    deformation = Double.parseDouble(LoadWeightJson.get("251").toString());
//                                    SimpleDateFormat format = new SimpleDateFormat("yyMMddHHmmss");
//                                    createTime = format.parse(jsonMap.get("datetime"));
//                                } catch (Exception ignored) {
//
//                                }
//
//
//                                SignCarEntity car = signCarService.getSignCarByClientId(clientId);
//
//                                if (car != null) {
//                                    SignCarLonLatEntity entity = new SignCarLonLatEntity();
//                                    if (!StringUtils.isEmpty(weight)) {
//                                        entity.setWeight(weight);
//                                        entity.setDeformation(deformation);
//                                        entity.setCreateTime(createTime);
//                                    }
//                                    entity.setCarId(car.getId());
//                                    entity.setSpeed(speed);
//                                    entity.setLat(latitude);
//                                    entity.setLon(longitude);
//                                    entity.setDataTime(new Date());
//                                    entity.setDirection(direction);
//                                    signCarService.updateCarLonLat(entity);
//                                }
//                            }
//                        }
//==============================================================================================================================================
                    }

                    @Override
                    public void deliveryComplete(IMqttDeliveryToken token) {
//                        System.out.println("deliveryComplete....");
                    }
                });
                client.connect(options);
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布
     *
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
        if (null == mTopic) {
//            logger.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topics 主题
     * @param qos    连接方式
     */
    public void subscribe(String[] topics, int qos) {
//        logger.info("==============开始订阅主题=========" + topics);
        try {
            MqttPushClient.getClient().subscribe(topics, new int[]{qos, qos});
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

